package com.yc.mqtt.handler.message;

import com.alibaba.fastjson.JSON;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.*;
import io.netty.util.CharsetUtil;

public class MqttPublishMessageHandler extends MqttBaseMessageHandler {
    @Override
    public void onMessage(ChannelHandlerContext channelHandlerContext, MqttMessage mqttMessage) {
        MqttPublishMessage mqttPublishMessage = (MqttPublishMessage) mqttMessage;
        MqttFixedHeader mqttFixedHeader = mqttMessage.fixedHeader();

        MqttPublishVariableHeader mqttPublishVariableHeader = (MqttPublishVariableHeader) mqttMessage.variableHeader();
        System.out.println(mqttPublishVariableHeader);
        System.out.println(JSON.toJSONString(mqttPublishVariableHeader.properties()));


        ByteBuf payload = mqttPublishMessage.payload();
        byte[] bytes = new byte[payload.readableBytes()];
        payload.readBytes(bytes);
        System.out.println(new String(bytes, CharsetUtil.UTF_8));

        MqttFixedHeader respMqttFixedHeader = new MqttFixedHeader(MqttMessageType.PUBACK, mqttFixedHeader.isDup(), MqttQoS.AT_MOST_ONCE, mqttFixedHeader.isRetain(), 0x02);
        MqttMessageIdVariableHeader mqttConnAckVariableHeader = MqttMessageIdVariableHeader.from(1);
        MqttPubAckMessage ack = new MqttPubAckMessage(respMqttFixedHeader, mqttConnAckVariableHeader);
        channelHandlerContext.channel().writeAndFlush(ack);


        sendQos1Msg(channelHandlerContext.channel(), mqttPublishVariableHeader.topicName(), mqttFixedHeader.isDup(), bytes, mqttPublishVariableHeader.packetId());
    }


    /**
     * 发送 qos1 类的消息
     */
    private void sendQos1Msg(Channel channel, String topic, boolean isDup, byte[] byteBuf, int messageId) {
        MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, isDup, MqttQoS.AT_LEAST_ONCE, false, 0);
        MqttPublishVariableHeader mqttPublishVariableHeader = new MqttPublishVariableHeader(topic, messageId);
        MqttPublishMessage mqttPublishMessage = new MqttPublishMessage(mqttFixedHeader, mqttPublishVariableHeader, Unpooled.wrappedBuffer(byteBuf));
        channel.writeAndFlush(mqttPublishMessage);
    }
}
